package com.yujinglin.maxmin;


import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.math.BigDecimal;

/**
 * 对于求极值来说，只能有一个Reducer（1个分区）
 */
public class MaxMinReducer extends Reducer<IntWritable, DoubleWritable, Text,DoubleWritable> {
    private double max;
    private  double min;



    @Override
    protected void reduce(IntWritable key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {

        this.max=Long.MIN_VALUE;
        this.min=Long.MAX_VALUE;

        String name="无效数据";

        if(key.get()>0&&key.get()<=1){
            name="500以内的短程航班";
        }else if(key.get()>1&&key.get()<=2){
            name="500以上的短程航班";
        }else if(key.get()>2&&key.get()<=3){
            name="1500以内的中程航班";
        }else if(key.get()>3&&key.get()<=4){
            name="1500以上的中程航班";
        }else if(key.get()>4&&key.get()<=5){
            name="远程航班";
        }else if(key.get()>5){
            name="全球航班";
        }


        // 因为是1个分区+key都是null，所以，收到的所有临时最大值，都在values
        for(DoubleWritable tmp : values){
                min= Math.min(min, tmp.get());

        }


        BigDecimal temp = new BigDecimal(min);
        double endMin = temp.setScale(2,BigDecimal.ROUND_HALF_UP).doubleValue();

        context.write(new Text(name),new DoubleWritable(endMin));
    }

//    @Override
//    protected void cleanup(Context context) throws IOException, InterruptedException {
//        System.out.println("Reducer的cleanup方法被执行了");
//        Text text = null;
//
//    }


}
